package com.weishao.learn;

import java.util.Properties;
import org.pentaho.di.core.KettleEnvironment;
import org.pentaho.di.core.SQLStatement;
import org.pentaho.di.core.database.Database;
import org.pentaho.di.core.database.DatabaseMeta;
import org.pentaho.di.core.row.RowMetaInterface;
import org.pentaho.di.trans.TransHopMeta;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.steps.tableinput.TableInputMeta;
import org.pentaho.di.trans.steps.tableoutput.TableOutputMeta;

public class KettleLearnApplication {

	public static void main(String[] args) throws Exception {
		KettleEnvironment.init();

		TransMeta transMeta = new TransMeta();
		transMeta.setName("ktrans");

		DatabaseMeta sourceDatabaseMeta = new DatabaseMeta("source", "mysql", "Native(JDBC)", "172.16.90.210",	"kettle_test", "3306", "tangyibo", "tangyibo");
		sourceDatabaseMeta.addOptions();
		sourceDatabaseMeta.setQuoteAllFields(true);
		sourceDatabaseMeta.setStreamingResults(true);
		sourceDatabaseMeta.setConnectSQL("SELECT 1");

		DatabaseMeta targetDatabaseMeta = new DatabaseMeta("target", "POSTGRESQL", "Native(JDBC)", "172.16.90.210",	"tangyibo", "5432", "postgres", "postgres");
		//DatabaseMeta targetDatabaseMeta = new DatabaseMeta("target", "Greenplum", "Native(JDBC)", "172.16.90.151",	"study", "5432", "study", "123456");
		//DatabaseMeta targetDatabaseMeta = new DatabaseMeta("target", "Oracle", "Native(JDBC)", "172.16.90.158",	"ORCL", "1521", "tangyibo", "tangyibo");
		//DatabaseMeta targetDatabaseMeta = new DatabaseMeta("target", "Mysql", "Native(JDBC)", "172.16.90.210",	"kettle_test", "3306", "tangyibo", "tangyibo");
		//DatabaseMeta targetDatabaseMeta = new DatabaseMeta("target", "MSSQLNATIVE", "Native(JDBC)", "172.16.90.166","hqtest", "1433", "hqtest", "123456");
		targetDatabaseMeta.setQuoteAllFields(true);
		if ("MSSQLNATIVE".equals(targetDatabaseMeta.getDatabaseInterface().getPluginId())) {
			Properties attributes = new Properties();
			attributes.put("MSSQLUseIntegratedSecurity", "false");// 只对sqlserver有效
			targetDatabaseMeta.setAttributes(attributes);
			targetDatabaseMeta.setConnectSQL("SELECT 1");
		}else if ("POSTGRESQL".equals(targetDatabaseMeta.getDatabaseInterface().getPluginId())) {
			targetDatabaseMeta.setConnectSQL("SELECT 1");
		}else {
			targetDatabaseMeta.setConnectSQL("SELECT 1 FROM DUAL");
		}

		transMeta.addDatabase(sourceDatabaseMeta);
		transMeta.addDatabase(targetDatabaseMeta);

		TableInputMeta t_input = new TableInputMeta();
		t_input.setDatabaseMeta(sourceDatabaseMeta);
		//String schemaTable = sourceDatabaseMeta.getQuotedSchemaTableCombination("kettle_test", "CJB");
		//t_input.setSQL(String.format("select * from %s", schemaTable));

		TableOutputMeta t_output = new TableOutputMeta();
		t_output.setDatabaseMeta(targetDatabaseMeta);
		t_output.setSchemaName("public");
		t_output.setTableName("new_table_name");

		StepMeta inputStepMeta = new StepMeta("tableInput", t_input);
		transMeta.addStep(inputStepMeta);

		StepMeta outputStepMeta = new StepMeta("tableOutput", t_output);
		transMeta.addStep(outputStepMeta);

		transMeta.addTransHop(new TransHopMeta(inputStepMeta, outputStepMeta));

		Database db = new Database(sourceDatabaseMeta);
		db.shareVariablesWith(transMeta);
		db.connect();
		//RowMetaInterface rowMeta = db.getQueryFields(t_input.getSQL(), true);
		RowMetaInterface rowMeta = db.getTableFields("v_test");
		db.disconnect();
		
		System.out.println(rowMeta.toStringMeta());
		
		SQLStatement statement = t_output.getSQLStatements(transMeta, outputStepMeta, rowMeta, null, false, null);
		if(statement.hasSQL()) {
			System.out.println(statement.getSQL());
		}else {
			System.out.println(statement.getError());
		}
	}

}
